package com.homedo.rxj;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;

public class RxjTest2 {
    static ObservableEmitter<Integer> curEmitter = null;

    public static void main(String[] args) {

        createObserver().doOnComplete(new Action() {
            @Override
            public void run() throws Exception {
                System.out.println("Completed");
            }
        }).doOnError(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                System.out.println("Error:" + throwable.getMessage());
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                System.out.println(integer);
            }
        });
        

        for (int i = 0; i < 10000; i++) {
            if (!curEmitter.isDisposed()) {
                curEmitter.onNext(i);
                if (i == 10000 - 1) {
                    curEmitter.onComplete();
                }
            }
        }
    }


    private static Observable<Integer> createObserver() {
        return Observable.create(
                new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        curEmitter = emitter;
                    }
                }
        );
    }
}
